package f8watermark;

import orderdata.OrderData;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class WmCus {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> scostr = env.socketTextStream("192.168.43.102", 7777);
        //端口读取数据内容映射为订单类
        scostr.map(new MapFunction<String, OrderData>() {
            @Override
            public OrderData map(String s) throws Exception {
                String[] arr = s.split(",");
                return new OrderData(
                        Long.parseLong(arr[0]),   // order_date (timestamp)
                        arr[1],                   // category
                        arr[2],                   // region
                        Integer.parseInt(arr[3]), // user_id
                        Integer.parseInt(arr[4]), // order_count
                        Double.parseDouble(arr[5])// total_amount
                );
            }
        }).assignTimestampsAndWatermarks(new MyWatermarkStrategy());
    }
    // 自定义水位线策略(时间戳+Watermark 生成器)
    public static class MyWatermarkStrategy implements WatermarkStrategy<OrderData> {

        @Override
        public TimestampAssigner<OrderData> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<OrderData>() {
                @Override
                public long extractTimestamp(OrderData orderData, long l) {
                    return orderData.order_date;//从数据中提取时间戳
                }
            };
        }

        @Override
        public WatermarkGenerator<OrderData> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new MyWatermarkGenerator();//自定义 Watermark 生成器
        }
    }
    //自定义 Watermark 生成器类，实现 WatermarkGenerator 接口
    public static class MyWatermarkGenerator implements WatermarkGenerator<OrderData> {

        //设置最大允许延迟时间，单位为毫秒
        private final long maxDelay = 5000;//5s
        //已处理的最大时间戳
        private long maxTsSeen = Long.MIN_VALUE + maxDelay + 1L;
        //每次接收到新的事件时，onEvent 方法会被调用
        @Override
        public void onEvent(OrderData orderData, long l, WatermarkOutput watermarkOutput) {//记录了事件流中遇到的最大时间戳
            //确保记录的是事件流中的最大时间戳
            maxTsSeen = Math.max(maxTsSeen, orderData.order_date);
        }
        //会在一定的时间间隔内周期性地调用。每次调用时，它会根据当前最大事件时间戳生成一个水印，印的时间为 maxTsSeen - maxDelay- 1L，
        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {//水即最大时间戳减去一个最大延迟（在这里是 5 秒）。
            //创建并发出一个新的水印，水印时间为已知的最大时间戳减去最大延迟时间
            watermarkOutput.emitWatermark(new Watermark(maxTsSeen-maxDelay- 1L));
        }
    }
}
